--------------------------sql任务说明----------------------------------
-- @author :liuxiaojie05
-- @descrbe :简历发布/更新数据 source=kafka b2c ods --->dwd
-- @udf jar包 : bigdata_realtime_flink_sql_udf-1.0-testlxj-jar-with-dependencies.jar
-- 下游使用：olap

--------------------------sql任务说明----------------------------------

-- 加载函数
CREATE FUNCTION ZpDWDResumePostUdtf as 'udtf.etl.dwd.ZpDWDResumePostUdtf';
create function SplitFunction AS 'udtf.SplitFunction';


-- source
create table ods_zp_resume_post(
log string
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_zhaopin_resumepost',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_resumepost-sql-prod',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_resumepost-OQB1K',
'format' = 'raw'
);

-- sink
create table dwd_zp_resume_post (
    id     BIGINT,
    oper   INT,
	step   int,
	whichMethod string,
	modifiedFields string,
	entity string,
	tjfrom string,
	platform INT,
	addTime BIGINT,
	sink_time string,
	e_time BIGINT
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_resume_post',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_resume_post-sql-prod',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_resume_post-QvyDC',
'format' = 'json'
);


-- sql-etl
insert into dwd_zp_resume_post
select pp.*
from (
    select regex_str
    from ods_zp_resume_post,LATERAL TABLE(SplitFunction(log,'\n'))
) tmp, LATERAL TABLE(ZpDWDResumePostUdtf(regex_str)) as pp;